<?php
/**
 * Created by PhpStorm.
 * User: BlueIdeaLirui
 * Date: 2019/11/11
 * Time: 10:13
 */

namespace business\bigdata\services\queues;

use business\base\caches\DataCache;
use business\bigdata\repositories\SupplierRepository;
use common\queue\QueueConstants;
use \business\base\services\queues\BaseService;
use common\utils\DateTimeUtils;

class SupplierProducerService extends BaseService
{
    /**
     * @var \iQueue\base\AbstractQueueAdapter
     */
    protected $queue;

    public function __construct()
    {
        parent::__construct();
        $this->queue = \yii::$app->iQueue->createClient(QueueConstants::BIG_DATA_SUPPLIER_INDEX_QUEUE);
    }

    /**
     * 增量方式更新时: 记录最近一次供应商索引查询的时间的key
     */
    const LATEST_SUPPLIER_QUERY_TIME_KEY = 'Last-Bigdata-Supplier-Query-Time';

    /**
     * 指定模块
     */
    const MODULE = 'supplier';

    /**
     * 全量更新供应商索引
     * @return mixed|void
     */
    public function full()
    {
        $supplierRepository = new SupplierRepository();
        $totalCount = $supplierRepository->getSupplierCountByDate();
        $loopCount = ceil($totalCount / self::MAX_COUNT_PER_QUERY);
        $lastSupplierId = "0";
        $dataCache = new DataCache();
        for ($i = 0; $i < $loopCount; $i++) {
            $changedSupplierIds = $supplierRepository->getAllSupplierId($lastSupplierId, self::MAX_COUNT_PER_QUERY);

            $lastSupplierId = $changedSupplierIds[count($changedSupplierIds) - 1];

            $values = implode(self::DELIMITER, $changedSupplierIds);
            $data = [
                'source' => 'bigdata',
                'values' => $values,
                'uuid' => $this->uuid(),
                'created_on' => DateTimeUtils::now()
            ];
            $cacheKey = $message = md5(serialize($data));

            $queueCacheName = self::MODULE . '-' . QueueConstants::QUEUE_CHACHE_NAME;

            // 如果缓存数据存在则先删除原数据
            if ($this->existInCache($queueCacheName, $cacheKey)) {
                $dataCache->delRecordByHash($queueCacheName, $cacheKey);
            }

            $dataCache->saveRecordByHash($queueCacheName, $cacheKey, $data);

            $this->queue->sendMsg($message, function ($msgBody) use ($loopCount, $i) {
                echo '大数据供应商-已插入队列比例: ' . ($i + 1) . '/' . $loopCount . PHP_EOL;
            });

            usleep(self::MICRO_SECOND);
        }
    }

    /**
     * 增量更新供应商索引
     * @return mixed|void
     */
    public function increment()
    {
        $supplierRepository = new SupplierRepository();
        $dataCache          = new DataCache();

        // 返回上一次查询时间
        list($startData, $endData) = $dataCache->getLastQueryTime(self::LATEST_SUPPLIER_QUERY_TIME_KEY);

        $totalCount = $supplierRepository->getSupplierCountByTimeSlot($startData, $endData);
        $loopCount = ceil($totalCount / self::MAX_COUNT_PER_QUERY);
        $lastSupplierId = "0";

        for ($i = 0; $i < $loopCount; $i++) {
            $changedSupplierIds = $supplierRepository->getChangedSupplierInfoByTimeSlot($startData, $endData,
                $lastSupplierId, self::MAX_COUNT_PER_QUERY);

            if (empty($changedSupplierIds)) {
                continue;
            }

            $lastSupplierId = $changedSupplierIds[count($changedSupplierIds) - 1];

            $values   = implode(self::DELIMITER, $changedSupplierIds);
            $data     = ['source' => 'bigdata', 'values' => $values, 'uuid' => $this->uuid(), 'created_on' => DateTimeUtils::now()];
            $cacheKey = $message = md5(serialize($data));

            $queueCacheName = self::MODULE . '-' . QueueConstants::QUEUE_CHACHE_NAME;

            // 如果缓存数据存在则先删除原数据
            $this->existInCache($queueCacheName, $cacheKey)
                && $dataCache->delRecordByHash($queueCacheName, $cacheKey);

            $dataCache->saveRecordByHash($queueCacheName, $cacheKey, $data);

            $this->queue->sendMsg($message, function ($msgBody) use($loopCount, $i) {
                echo '大数据供应商-已插入队列比例: ' . ($i + 1) . '/' . $loopCount . PHP_EOL;
            });

            usleep(self::MICRO_SECOND);
        }
    }

    /**
     * 更新单个供应商索引
     * @param $supplierId
     */
    public function single($supplierId)
    {
        if (empty($supplierId)) {
            echo '大数据供应商-单个供应商信息更新失败: 供应商ID不能为空' . PHP_EOL;
        }
        $supplierRepository = new SupplierRepository();
        $dataCache          = new DataCache();

        $exist = $supplierRepository->existInSupplierById($supplierId);
        if ($exist === false) {
            echo '大数据供应商-单个供应商信息更新失败: 供应商信息不存在，对应的供应商id为：' . $supplierId . PHP_EOL;
        }

        $data = [
            'source' => 'bigdata',
            'values' => $supplierId,
            'uuid' => $this->uuid(),
            'created_on' => DateTimeUtils::now()
        ];
        $cacheKey = $message = md5(serialize($data));
        $queueCacheName = self::MODULE . '-' . QueueConstants::QUEUE_CHACHE_NAME;

        // 如果缓存数据存在则先删除原数据
        if ($this->existInCache($queueCacheName, $cacheKey)) {
            $dataCache->delRecordByHash($queueCacheName, $cacheKey);
        }
        $dataCache->saveRecordByHash($queueCacheName, $cacheKey, $data);

        $this->queue->sendMsg($message, function ($msgBody) use($supplierId) {
            echo '大数据供应商' . $supplierId . '-已插入队列: ' . PHP_EOL;
        });
    }

    /**
     * 重发机制
     */
    public function resend()
    {
        $this->processResend(
            QueueConstants::BIG_DATA_SUPPLIER_INDEX_QUEUE
            , self::MODULE
            , '大数据供应商'
        );
    }

    /**
     * 清除已经消费过的队列数据
     * @throws \yii\base\InvalidConfigException
     */
    public function clearConsumerQueue()
    {
        $this->processclearConsumerQueue(self::MODULE);
    }
}
